package org.lds.hotkey.client.autoconfigure.hotkey.report;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.lds.hotkey.common.autoconfigure.spring.SpringUtil;
import org.lds.hotkey.client.holder.WorkerHolder;
import org.lds.hotkey.common.model.ReportRuleModel;
import org.lds.hotkey.common.model.WorkerModel;

import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.stream.Collectors;

@Slf4j
public class ReportTask implements Runnable {

    private final BlockingQueue<ReportRuleModel> queue;
    private WorkerService workerService = SpringUtil.getBean(WorkerService.class);

    public ReportTask(BlockingQueue<ReportRuleModel> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            report();
        } catch (Throwable e) {
            log.warn("上报失败", e);
        }
    }

    private void report() {
        // 不会阻塞
        List<ReportRuleModel> reportRuleModels = getReportData();
        if (CollectionUtils.isEmpty(reportRuleModels)) return;

        // 发送到worker
        pushWorker(reportRuleModels);
    }

    private List<ReportRuleModel> getReportData() {
        List<ReportRuleModel> reportRuleModels = new ArrayList<>();
        queue.drainTo(reportRuleModels);
        return reportRuleModels;
    }

    private void pushWorker(List<ReportRuleModel> reportRuleModels) {
        // 基于key分组
        Map<String, List<ReportRuleModel>> keyGroupMap = reportRuleModels.stream().collect(Collectors.groupingBy(reportRuleModel -> reportRuleModel.getKey()));
        Map<String, ReportRuleModel> keyReportRuleModelMap = reportRuleModelMap(keyGroupMap);

        // 根据key查找worker
        Map<String, ReportRuleModel> workerReportRuleModelMap = workerReportRuleModelMap(keyReportRuleModelMap);

        // 根据worker异步推送数据
        workerService.asyncPush(workerReportRuleModelMap);
    }

    private Map<String, ReportRuleModel> reportRuleModelMap(Map<String, List<ReportRuleModel>> keyGroupMap) {
        Map<String, ReportRuleModel> reportRuleModelMap = new HashMap<>();
        keyGroupMap.forEach((key, reportRuleModels) -> {
            if (CollectionUtils.isNotEmpty(reportRuleModels)) {
                ReportRuleModel reportRuleModel = reportRuleModels.get(0);
                reportRuleModel.setCount(reportRuleModels.size());
                reportRuleModelMap.put(key, reportRuleModel);
            }
        });
        return reportRuleModelMap;
    }

    /**
     * key: worker(ip:port)
     * value:
     *   key: key
     *   value: reportRuleModels
     * @param keyReportRuleModelMap
     * @return
     */
    private Map<String, ReportRuleModel> workerReportRuleModelMap(Map<String, ReportRuleModel> keyReportRuleModelMap) {
        Map<String, ReportRuleModel> workerReportRuleModelMap = new HashMap<>();
        keyReportRuleModelMap.forEach((key, reportRuleModel) -> {
            WorkerModel workerModel = WorkerHolder.getWorkerModel(key);
            if (Objects.nonNull(workerModel)) {
                workerReportRuleModelMap.put(workerModel.buildAddress(), reportRuleModel);
            }
        });
        return workerReportRuleModelMap;
    }

}
